package com.chis.activemq;

import com.zendesk.maxwell.MaxwellContext;
import com.zendesk.maxwell.producer.AbstractProducer;
import com.zendesk.maxwell.row.RowMap;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;


public class ActiveMqProducer extends AbstractProducer {

    /* 日志 */
    private static final Logger LOGGER = LoggerFactory.getLogger(ActiveMqProducer.class);
    /* 获取配置 */
    //ActiveMq 的默认用户名
    private final String USERNAME;
    //ActiveMq 的默认登录密码
    private final String PASSWORD;
    //ActiveMQ 的链接地址
    private final String BROKEN_URL;
    private final String QUEUE_NAME;
    //
    private ConnectionFactory connectionFactory;
    private Connection connection;
    private Session session;


    public ActiveMqProducer(MaxwellContext context) {
        super(context);
        // this property would be 'custom_producer.header_format' in config.properties
        USERNAME = context.getConfig().customProducerProperties.getProperty("activemq_username");
        PASSWORD = context.getConfig().customProducerProperties.getProperty("activemq_password");
        BROKEN_URL = context.getConfig().customProducerProperties.getProperty("activemq_brokenurl");
        QUEUE_NAME = context.getConfig().customProducerProperties.getProperty("activemq_queuename");
        //
        try {
            this.connectionFactory =  new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
            this.connection  = connectionFactory.createConnection();
//            this.session = connection.createSession(true,Session.SESSION_TRANSACTED);
            this.session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        } catch (Exception e){
            LOGGER.error(e.getMessage());
            throw new RuntimeException(e);
        }
    }

    @Override
    public void push(RowMap r) throws Exception    {
        // filtering out DDL and heartbeat rows
        if(!r.shouldOutput(outputConfig)) {
            // though not strictly necessary (as skipping has no side effects), we store our position,
            // so maxwell won't have to "re-skip" this position if crashing and restarting.
            context.setPosition(r.getNextPosition());
            return;
        }

        String tmp = r.toJSON(outputConfig);
        LOGGER.info(tmp);
        TextMessage textMessage = session.createTextMessage(tmp);
        for(String queuename : QUEUE_NAME.split("#")){
            MessageProducer messageProducer = session.createProducer(session.createQueue(queuename));
            messageProducer.send(textMessage);
        }

        if(r.isTXCommit()) {
            context.setPosition(r.getNextPosition());
        }
    }



}
